package com.shujia.tour

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object CityTouristWideApp extends Logging {
  def main(args: Array[String]): Unit = {

    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id: String = args(0)
    val month_id: String = day_id.substring(0, 6)

    log.info(s"当前天时间分区为：$day_id")
    log.info(s"当前月时间分区为：$month_id")

    val spark: SparkSession = SparkSession
      .builder()
      .appName("StayPointApp")
      .config("spark.sql.shuffle.partitions", "1")
      .enableHiveSupport() //使用hive元数据
      .getOrCreate()

    /**
     * 关联事实表和维度表构建宽表
     *
     * 关联市游客表，用户画像表，行政区配置表
     *
     */

    /**
     * d_city_name string comment '旅游目的地市名'
     * ,o_city_name string comment '旅游来源地地市名'
     * ,o_province_name string comment '旅游来源地省名'
     * ,number_attr string comment '号码归属地'
     * ,d_distance_section string comment '出游距离'
     * ,d_stay_time string comment '停留时间按小时'
     * ,gender string comment '性别'
     * ,trmnl_brand string comment '终端品牌'
     * ,pckg_price int comment '套餐'
     * ,conpot int comment '消费潜力'
     * ,age int comment '年龄'
     *
     */

    val wideDF: DataFrame = spark.sql(
      s"""
         |
         |SELECT
         |    t5.d_city_name,
         |    t5.o_city_name,
         |    t5.o_province_name,
         |    t6.number_attr,
         |    case
         |        when  t5.d_max_distance > 10 and  t5.d_max_distance<=50 then "10-50"
         |        when  t5.d_max_distance > 50 and  t5.d_max_distance<=80 then "50-80"
         |        when  t5.d_max_distance > 80 and  t5.d_max_distance<=120 then "80-120"
         |        else "120~"
         |    end as d_distance_section,
         |    case
         |        when t5.d_stay_time>3 and t5.d_stay_time<5 then "3-5"
         |        when t5.d_stay_time>5 and t5.d_stay_time<8 then "5-8"
         |        when t5.d_stay_time>8 and t5.d_stay_time<12 then "8-12"
         |        when t5.d_stay_time>12 and t5.d_stay_time<20 then "12-20"
         |        else "20~"
         |    end as d_stay_time,
         |    t6.gender,
         |    t6.trmnl_brand,
         |    t6.packg,
         |    t6.conpot,
         |    t6.age
         |
         |FROM
         |    (
         |    SELECT
         |        /*+ BROADCAST(t4) */
         |        t3.mdn,
         |        t4.city_name AS d_city_name,
         |        t3.o_city_name,
         |        t3.o_province_name,
         |        t3.d_max_distance,
         |        t3.d_stay_time
         |
         |    FROM
         |        (
         |        SELECT
         |            /*+ BROADCAST(t2) */
         |            t1.mdn,
         |            t1.d_city_id,
         |            t1.d_stay_time,
         |            t1.d_max_distance,
         |            t2.prov_name AS o_province_name,
         |            t2.city_name AS o_city_name
         |
         |        FROM
         |            (
         |            SELECT
         |                mdn,
         |                source_county_id,
         |                d_city_id,
         |                d_stay_time,
         |                d_max_distance
         |
         |            FROM
         |                dal_tour.dal_tour_city_tourist_msk_d
         |
         |            WHERE day_id = $day_id
         |            ) AS t1
         |
         |            JOIN dim.dim_admincode AS t2 ON t1.source_county_id = t2.county_id
         |        ) AS t3
         |
         |        JOIN (
         |            SELECT
         |                DISTINCT
         |                city_id,
         |                city_name
         |            FROM
         |            dim.dim_admincode
         |            ) AS t4 ON t3.d_city_id = t4.city_id
         |
         |    ) AS t5
         |
         |    JOIN ( SELECT * FROM dim.dim_usertag_msk_m WHERE month_id = $month_id ) AS t6 ON t5.mdn = t6.mdn
         |
         |""".stripMargin)

    //保存数据
    wideDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_city_wide_msk_d/day_id=$day_id")

    //增加分区
    spark.sql(s"alter table dal_tour.dal_tour_city_wide_msk_d add if not exists partition(day_id='$day_id')")

  }
}
